package disnetwork;

import org.apache.commons.lang3.StringUtils;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author 龙小虬
 * @date 2021/5/6 22:28
 */
public class MQServer {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingQueue<String> mqServer = new LinkedBlockingQueue<>();
        // 创建生产者线程
        Thread producerThread = new Thread(() -> {
            for (int i = 1; i <= 1000; i++) {
                String msg = "msg:" + i;
                System.out.println(Thread.currentThread().getName() + "," + msg);
                mqServer.offer(msg);
            }
        }, "生产者线程：");
        producerThread.start();
        producerThread.join();
        System.out.println("生产者将数据放入到MQ中完成...");

        Thread consumerThread1 = new Thread(() -> {
            for (; ; ) {
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {

                }
                String poll = mqServer.poll();
                if (!StringUtils.isEmpty(poll)) {
                    System.out.println(Thread.currentThread().getName() + "," + poll);
                }
            }
        }, "消费者线程1：");
        consumerThread1.start();
        Thread consumerThread2 = new Thread(() -> {
            for (; ; ) {
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {

                }
                String poll = mqServer.poll();
                if (!StringUtils.isEmpty(poll)) {
                    System.out.println(Thread.currentThread().getName() + "," + poll);
                }
            }
        }, "消费者线程2：");
        consumerThread2.start();
        Thread consumerThread3 = new Thread(() -> {
            for (; ; ) {
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {

                }
                String poll = mqServer.poll();
                if (!StringUtils.isEmpty(poll)) {
                    System.out.println(Thread.currentThread().getName() + "," + poll);
                }
            }
        }, "消费者线程3：");
        consumerThread3.start();
    }
}
